package org.sunyaxing.transflow.transflowapp.controllers;

import cn.hutool.core.date.DateUtil;
import jakarta.websocket.OnClose;
import jakarta.websocket.OnMessage;
import jakarta.websocket.OnOpen;
import jakarta.websocket.Session;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.helpers.MessageFormatter;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@ServerEndpoint(value = "/transflow/job/logs/{jobId}")
public class JobLogsWs {
    public static final Map<String, Map<String, Session>> SESSION_MAP = new HashMap<>();
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    @OnOpen
    public void onOpen(Session session, @PathParam("jobId") String jobId) {
        log.info("监听日志，在线{}", ONLINE_COUNT.incrementAndGet());
        if (!SESSION_MAP.containsKey(jobId)) {
            Map<String, Session> sessions = new HashMap<>();
            sessions.put(session.getId(), session);
            SESSION_MAP.put(jobId, sessions);
        } else {
            SESSION_MAP.get(jobId).put(session.getId(), session);
        }
    }

    @OnClose
    public void onClose(Session session, @PathParam("jobId") String jobId) {
        if (!SESSION_MAP.containsKey(jobId)) {
        } else {
            if (SESSION_MAP.get(jobId).isEmpty()) {
            } else {
                try {
                    log.info("当前在线{}", ONLINE_COUNT.decrementAndGet());
                    session.close();
                } catch (Exception e) {
                    log.error("close session error", e);
                } finally {
                    SESSION_MAP.get(jobId).remove(session.getId());
                }
            }
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
    }

    public static void sendMessage(String jobId, String message, Object... argumentArray) {
        log.info(message, argumentArray);
        // 薛定谔的猫，观测状态下会坍缩到低吞吐状态
        if (ONLINE_COUNT.get() <= 0) return;
        try {
            if (SESSION_MAP.containsKey(jobId)) {
                String dateTime = "【" + DateUtil.now() + "】";
                String msg = MessageFormatter.arrayFormat(message, argumentArray).getMessage();
                synchronized (SESSION_MAP) {
                    SESSION_MAP.get(jobId).forEach((key, session) -> {
                        try {
                            session.getBasicRemote().sendText(dateTime + msg);
                        } catch (Exception e) {
                            log.error("send message error", e);
                            try {
                                session.close();
                            } catch (Exception e2) {
                                log.error("send message error", e2);
                            }
                        }
                    });
                }
            }
        } catch (Exception e) {
            log.error("send message error", e);
        }
    }
}
